#include "include/zookeeperutil.h"
#include "MrpcLog.h"
#include "MrpcApplication.h"
#include <iostream>
#include <cstring>
#include <chrono>

Zkclient::Zkclient() : m_zhandle(nullptr) {
    sem_init(&m_connect_sem, 0, 0);
}

Zkclient::~Zkclient() {
    if (m_zhandle != nullptr) {
        zookeeper_close(m_zhandle);
    }
    sem_destroy(&m_connect_sem);
}

// 全局监视器回调
void Zkclient::global_watcher(zhandle_t *zh, int type, 
                             int state, const char *path, void *watcherCtx) {
    if (type == ZOO_SESSION_EVENT) {
        if (state == ZOO_CONNECTED_STATE) {
            Zkclient* client = static_cast<Zkclient*>(watcherCtx);
            sem_post(&client->m_connect_sem);
            LOG_INFO("ZooKeeper connected successfully");
        } else if (state == ZOO_EXPIRED_SESSION_STATE) {
            LOG_ERROR("ZooKeeper session expired");
        }
    }
}

void Zkclient::Start() {
    std::string host = MrpcApplication::GetInstance().GetConfig().Load("zookeeperip");
    std::string port = MrpcApplication::GetInstance().GetConfig().Load("zookeeperport");
    std::string connstr = host + ":" + port;

    // 设置 ZooKeeper 日志级别
    zoo_set_debug_level(ZOO_LOG_LEVEL_WARN);

    // 初始化 ZooKeeper 连接
    m_zhandle = zookeeper_init(connstr.c_str(), global_watcher, 
                              30000, nullptr, this, 0);
    if (m_zhandle == nullptr) {
        LOG_ERROR("zookeeper_init failed");
        exit(EXIT_FAILURE);
    }

    // 等待连接建立
    struct timespec ts;
    clock_gettime(CLOCK_REALTIME, &ts);
    ts.tv_sec += 10; // 10秒超时
    
    if (sem_timedwait(&m_connect_sem, &ts) == -1) {
        LOG_ERROR("Connect to ZooKeeper timeout");
        zookeeper_close(m_zhandle);
        m_zhandle = nullptr;
        exit(EXIT_FAILURE);
    }
    
    LOG_INFO("ZooKeeper connection established successfully");
}

bool Zkclient::Exists(const std::string& path) {
    // 使用 promise-future 处理异步结果
    auto promise = std::make_shared<std::promise<bool>>();
    auto future = promise->get_future();

    // 异步检查节点是否存在
    int rc = zoo_aexists(m_zhandle, path.c_str(), 0, 
        [](int rc, const struct Stat *stat, const void *data) {
            auto promise_ptr = static_cast<std::shared_ptr<std::promise<bool>>*>(
                const_cast<void*>(data));
            promise_ptr->get()->set_value(rc == ZOK);
            delete promise_ptr;
        }, 
        new std::shared_ptr<std::promise<bool>>(promise)
    );

    if (rc != ZOK) {
        LOG_ERROR("Failed to initiate exists check: %s", zerror(rc));
        return false;
    }

    // 等待异步操作完成
    if (future.wait_for(std::chrono::seconds(5)) == std::future_status::timeout) {
        LOG_ERROR("Exists operation timeout");
        return false;
    }

    return future.get();
}

bool Zkclient::Create(const std::string& path, const std::string& data, int state) {
    // 首先检查节点是否已存在
    if (Exists(path)) {
        LOG_INFO("Node %s already exists", path.c_str());
        return true;
    }

    // 使用 promise-future 处理异步结果
    auto promise = std::make_shared<std::promise<bool>>();
    auto future = promise->get_future();

    // 异步创建节点
    int rc = zoo_acreate(m_zhandle, path.c_str(), data.c_str(), data.length(),
        &ZOO_OPEN_ACL_UNSAFE, state,
        [](int rc, const char *value, const void *data) {
            auto promise_ptr = static_cast<std::shared_ptr<std::promise<bool>>*>(
                const_cast<void*>(data));
            promise_ptr->get()->set_value(rc == ZOK);
            
            if (rc == ZOK) {
                LOG_INFO("Node created successfully: %s", value);
            } else {
                LOG_ERROR("Failed to create node: %s", zerror(rc));
            }
            
            delete promise_ptr;
        },
        new std::shared_ptr<std::promise<bool>>(promise)
    );

    if (rc != ZOK) {
        LOG_ERROR("Failed to initiate node creation: %s", zerror(rc));
        return false;
    }

    // 等待异步操作完成
    if (future.wait_for(std::chrono::seconds(5)) == std::future_status::timeout) {
        LOG_ERROR("Create operation timeout");
        return false;
    }

    return future.get();
}

std::string Zkclient::GetData(const std::string& path) {
    // 使用 promise-future 处理异步结果
    auto promise = std::make_shared<std::promise<std::string>>();
    auto future = promise->get_future();

    // 异步获取数据
    int rc = zoo_aget(m_zhandle, path.c_str(), 0,
        [](int rc, const char *value, int value_len, 
           const struct Stat *stat, const void *data) {
            auto promise_ptr = static_cast<std::shared_ptr<std::promise<std::string>>*>(
                const_cast<void*>(data));
            
            if (rc == ZOK && value != nullptr) {
                promise_ptr->get()->set_value(std::string(value, value_len));
                LOG_INFO("Get data successfully");
            } else {
                LOG_ERROR("Failed to get data: %s", zerror(rc));
                promise_ptr->get()->set_value("");
            }
            
            delete promise_ptr;
        },
        new std::shared_ptr<std::promise<std::string>>(promise)
    );

    if (rc != ZOK) {
        LOG_ERROR("Failed to initiate data retrieval: %s", zerror(rc));
        return "";
    }

    // 等待异步操作完成
    if (future.wait_for(std::chrono::seconds(5)) == std::future_status::timeout) {
        LOG_ERROR("GetData operation timeout");
        return "";
    }

    return future.get();
}